package job.relevance.anchor_weight;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import mapred.base.TextMapper;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

public class RLV_AW_Find_Mapper extends TextMapper {

    private static Map<String, List<String>> map = null;

    @Override
    public void map(Text key, Text value) throws IOException {
        String k = key.toString();
        int pos = k.indexOf("#");
        String url = k.substring(0, pos);
        /**
         * 
         */
        url = url.toLowerCase();
        /**
         * 
         */
        if (map.containsKey(url)) {
            String type = k.substring(pos + 1, pos + 2);
            k = k.substring(pos + 2).replaceAll("/\\*\\*\\*/", "\t");
            List<String> l = new ArrayList<String>();
            for (String s : k.split("\t")) {
                l.add(s);
            }
            String weight = value.toString().replaceAll("\t", "\1");

            for (String header : map.get(url)) {
                output(header, type, weight, l);
            }
        }
    }

    private void output(String header, String type, String weight,
            List<String> text) throws IOException {
        List<String> query = Arrays.asList(header.substring(0,
                header.indexOf("\t")).split(" "));
        String anchor = "\2" + type + "\1" + weight + "\1" + text.size();
        for (String q : query) {
            List<Integer> poss = findPosition(text, q);
            if (poss.size() > 0) {
                anchor += "\1" + q + "\t" + poss.size();
                for (int p : poss) {
                    anchor += "\t" + p + "\t" + 0;
                }
            }
        }
        this.output(new Text(header.replaceAll("\t", "/***/")),
                new Text(anchor));
    }

    private List<Integer> findPosition(List<String> text, String s) {
        List<Integer> ret = new ArrayList<Integer>();
        int pos = text.indexOf(s);
        while (pos >= 0) {
            ret.add(pos);
            text.set(pos, "(NULL)");
            pos = text.indexOf(s);
        }
        return ret;
    }

    @Override
    public void setUp() {
        map = new HashMap<String, List<String>>();

        String list = this.getConfig("list");

        try {
            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(list))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                if (!line.trim().isEmpty()) {
                    int pos = line.indexOf("\t");
                    String k = line.substring(0, pos);
                    /**
                     * 
                     */
                    k = k.toLowerCase();
                    /**
                     * 
                     */
                    String v = line.substring(pos + 1);
                    if (!map.containsKey(k)) {
                        map.put(k, new ArrayList<String>());
                    }
                    map.get(k).add(v);
                }
            }
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(255);
        }
    }

    @Override
    public void tearDown() throws IOException {
    }

}
